package com.hbase.tools;

import java.io.IOException;

import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Delete;
import org.apache.hadoop.hbase.client.Get;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Result;
import org.apache.hadoop.hbase.client.ResultScanner;
import org.apache.hadoop.hbase.client.Scan;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.filter.CompareFilter.CompareOp;
import org.apache.hadoop.hbase.filter.Filter;
import org.apache.hadoop.hbase.filter.RegexStringComparator;
import org.apache.hadoop.hbase.filter.RowFilter;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/*
 * 1.数据的增删改查,需要使用的是Table对象
 *
 * 2.Put	:	代表对单行数据的put操作
 *
 * 3.在hbase中，操作的数据都是以byte[]形式存在，需要把常用的数据类型转为byte[]
 * 			hbase提供了Bytes工具类:
 * 				Bytes.toBytes(x):基本数据类型转byte[]
 * 				Bytes.toXxx(x):从byte[]类型转Xxx类型
 *
 * 4.Get	:	代表对单行数据的get操作
 *
 * 5.Result	:	scan或get的单行的所有记录！
 *
 * 6.Cell	:	代表一个单元格！
 * 			hbase提供了CellUtil.clonexxx(Cell)，来获取cell中的列族、列名和值
 *
 * 7.ResultScanner	:	多行Result的集合
 */

public class DataUtil {

	// 用来记录一些信息
	private static Logger logger = LoggerFactory.getLogger(DataUtil.class);

	// 先获取表的Table对象
	public static Table getTable(Connection conn, String tableName, String nsname) throws IOException {

		// 验证表是否存在
		if (!TableUtil.ifTableExists(conn, tableName, nsname)) {
			logger.error(tableName + "表不存在！");
			return null;
		}

		// 获取表名
		TableName tn = TableUtil.checkTableName(tableName, nsname);

		Table table = conn.getTable(tn);

		return table;

	}

	// put
	public static void put(Connection conn, String tableName, String nsname, String rowkey,
						   String cf, String cq,String value) throws IOException {

		Table table = getTable(conn, tableName, nsname);

		if (table == null) {
			return;
		}

		// 创建Put对象
		Put put = new Put(Bytes.toBytes(rowkey));

		// 向rowkey中添加一个列族
		// 添加多个列族时可以put.addColumn().addColumn();
		put.addColumn(Bytes.toBytes(cf), Bytes.toBytes(cq) , Bytes.toBytes(value));

		table.put(put);
		System.out.println("成功添加一条数据！");

		table.close();

	}

	// get 表名 rowkey
	public static void get(Connection conn, String tableName, String nsname, String rowkey) throws IOException {

		Table table = getTable(conn, tableName, nsname);

		if (table == null) {
			return;
		}

		// 创建Get对象
		Get get = new Get(Bytes.toBytes(rowkey));

		// 设置单行查询的详细信息
		// 设置查哪个列
//		get.addColumn(family, qualifier)
		// 设置查哪个列族
//		get.addFamily(family)
		// 只查某个时间戳的数据
//		get.setTimeStamp(timestamp)
		//设置返回的version
		get.setMaxVersions();

		Result result = table.get(get);

		// 遍历结果
		parseResult(result);

		table.close();

	}

	// 遍历Result
	public static void parseResult(Result result) {

		if(result != null) {

			Cell[] rawCells = result.rawCells();

			for (Cell cell : rawCells) {

				System.out.println(
						String.format("行键：%s，列族：%s，列：%s，值：%s，时间戳：%d",
								Bytes.toString(CellUtil.cloneRow(cell)),
								Bytes.toString(CellUtil.cloneFamily(cell)),
								Bytes.toString(CellUtil.cloneQualifier(cell)),
								Bytes.toString(CellUtil.cloneValue(cell)),
								cell.getTimestamp()
						)
				);

			}

		}

	}

	// scan 表名 ,{STARTROW => X,STOPROW => X,LIMIT => 1}
	public static void scan(Connection conn, String tableName, String nsname) throws IOException {

		Table table = getTable(conn, tableName, nsname);

		if (table == null) {
			return;
		}

		// 构建Scan对象
		Scan scan = new Scan();
		Filter filter = new RowFilter(CompareOp.EQUAL,new RegexStringComparator("1002\\|.*\\|20201231"));
		scan.setMaxVersions();
		scan.setFilter(filter);


		// 设置扫描的起始行
//		scan.setStartRow(startRow);

		// 结果集扫描器
		ResultScanner scanner = table.getScanner(scan);

		for (Result result : scanner) {

			parseResult(result);

		}

		table.close();

	}

	// delete 表名,rowkey,[列族].[列],[时间戳]
	public static void delete(Connection conn, String tableName, String nsname, String rowkey) throws IOException {

		Table table = getTable(conn, tableName, nsname);

		if (table == null) {
			return;
		}

		// 构建Delete对象
		Delete delete = new Delete(Bytes.toBytes(rowkey));

		// 设置delete时的参数
		// 删除某个具体的值，为此列的最新Cell
		// 添加一条type=Delete的标记，只能删除最新的一条记录
		// 如果有历史版本的记录，无法删除
//		delete.addColumn(Bytes.toBytes("cf1"), Bytes.toBytes("age"));

		// 删除指定列的所有版本的数据
		// 添加一条type=DeleteColumn的标记
		delete.addColumns(Bytes.toBytes("cf1"), Bytes.toBytes("age"));

		// 删除指定列族的数据
		delete.addFamily(Bytes.toBytes("cf2"));

		table.delete(delete);

		table.close();

	}

}
